package com.enjoy.kafka.hellokafka;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;


import org.apache.kafka.clients.producer.*;
import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * 生产者
 */
public class KafkaProducerTest {

    static Logger logger = LoggerFactory.getLogger("enjoy");

    @Test
    public void test1() {
        Properties properties = new Properties();
        properties.put("metadata.broker.list", "192.168.126.128:9092");
        properties.put("key.serializer.class", "kafka.serializer.StringEncoder");
        properties.put("partitioner.class", "hai.mq.kafka.HaiPartitioner");

        Producer<String, byte[]> producer = new Producer<>(new kafka.producer.ProducerConfig(properties));

        producer.send(new KeyedMessage<>("hello-topic", null, "word".getBytes()));

        logger.info("消息发送完了");
    }

    @Test
    public void test2() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        // 机器的ip
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.126.128:9092");
        /*
         * acks=0,表示producer不需要等待任何broker确认收到消息的回复，就可以继续发送下一条消息。性能最高，但是最容易丢数据
         * acks=1,至少要等待leader已经成功将数据写入本地log，但是不需要等待所有follower是否成功写入。就可以继续发送下一条消息。这种情况下，如果follower没有成功备份数据，而此时leader又挂掉，则消息会丢失
         * acks=-1或all，需要等待min.insync.replicas(默认为1,推荐配置大于等于2)这个参数配置的副本个数都成功写入日志，这种策略只要有一个备份存活就不会丢失数据。这是最强的数据保证。除非是金融级别，或者跟钱打交道的场景才会使用这个配置
         */
        properties.put(ProducerConfig.ACKS_CONFIG, "all");
        /*
        发送失败会重试，默认重试间隔100ms，重试能保证消息发送的可靠性，但是也可能造成消息重复发送，比如网络抖动，所以需要在消费端做幂等性处理
         */
        properties.put(ProducerConfig.RETRIES_CONFIG, 3);
        /*
        重试间隔设置
         */
        properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 300);
        /*
        设置发送消息的本地缓冲区，如果设置了该缓冲区，消息会先发送到本地缓冲区，可以提高消息的发送性能，默认值是33554432,即32m
         */
        properties.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);
        /*
        kafka本地线程会从缓冲区取数据，批量发送到broker设置批量发送的大小，默认值是16384，即16kb，就是说一个batch满了就发送出去
         */
        properties.put(ProducerConfig.BATCH_SIZE_CONFIG, 16384);
        /*
        默认值0，意思是消息必须立即被发送，但是这样会影响性能
        一般设置10毫秒左右，就是这个消息发送完后会进入本地一个batch，如果10毫秒内，这个batch满了16kb就会随batch一起被发出
        如果10毫秒内，batch没满，那么也必须把消息发送出去，不能让消息的发送延迟时间太长
         */
        properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
        // 序列化
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 10; i++) {
            // 未指定发送分区，具体发送分区计算公式，hash(key)%partitionNum
//            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("hello-topic", "eqweeqd", "helloworld");
            // 指定发送分区
            ProducerRecord<String, String> producerRecord = new ProducerRecord<>("hello-topic", 0, "dasda", "dasd");
            // 等待消息发送成功的同步阻塞方法
            RecordMetadata recordMetadata = producer.send(producerRecord).get();
            logger.info("同步发送!" + "topic:" + recordMetadata.topic() + ",partition:" + recordMetadata.partition() + ",offset:" + recordMetadata.offset());

            // 异步回调方式发送消息
//            producer.send(producerRecord, new Callback() {
//                @Override
//                public void onCompletion(RecordMetadata metadata, Exception exception) {
//                    if (exception != null) {
//                        logger.error("发送消息失败");
//                    }
//                    if (metadata != null) {
//                        logger.info("同步发送!" + "topic:" + recordMetadata.topic() + ",partition:" + recordMetadata.partition() + ",offset:" + recordMetadata.offset());
//                    }
//                }
//            });

        }
        producer.close();
    }
}
